/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.nifi.processors.huawei.obs;

import com.obs.services.ObsClient;
import com.obs.services.model.GetObjectMetadataRequest;
import com.obs.services.model.ListVersionsResult;
import com.obs.services.model.ObjectMetadata;
import com.obs.services.model.VersionOrDeleteMarker;
import org.apache.nifi.annotation.behavior.*;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.state.Scope;
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.util.list.ListableEntity;
import org.apache.nifi.processor.util.list.ListedEntityTracker;
import org.apache.nifi.processors.huawei.obs.model.OBSObjectBucketLister;
import org.apache.nifi.processors.huawei.obs.model.OBSRecord;
import org.apache.nifi.processors.huawei.obs.model.ObsBucketLister;
import org.apache.nifi.serialization.record.RecordSchema;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;

import static org.apache.nifi.processors.huawei.obs.Constants.*;
import static org.apache.nifi.processors.huawei.common.PropertyDescriptors.*;

@PrimaryNodeOnly
@TriggerSerially
@TriggerWhenEmpty
@InputRequirement(Requirement.INPUT_FORBIDDEN)
@Tags({"Huawei", "OBS", "list"})
@CapabilityDescription("Retrieves a listing of objects from an obs bucket. For each object that is listed, creates a FlowFile that represents "
        + "the object so that it can be fetched in conjunction with FetchOBSObject. This Processor is designed to run on Primary Node only "
        + "in a cluster. If the primary node changes, the new Primary Node will pick up where the previous node left off without duplicating "
        + "all of the data.")
@Stateful(scopes = Scope.CLUSTER, description = "After performing a listing of keys, the timestamp of the newest key is stored, "
        + "along with the keys that share that same timestamp. This allows the Processor to list only keys that have been added or modified after "
        + "this date the next time that the Processor is run. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary "
        + "Node is selected, the new node can pick up where the previous node left off, without duplicating the data.")
@WritesAttributes({
        @WritesAttribute(attribute = OBS_BUCKET, description = "The name of the OBS bucket"),
        @WritesAttribute(attribute = "filename", description = "The name of the file"),
        @WritesAttribute(attribute = OBS_E_TAG, description = "The ETag that can be used to see if the file has changed"),
        @WritesAttribute(attribute = OBS_OWNER, description = "Object owner"),
        @WritesAttribute(attribute = OBS_IS_LATEST, description = "A boolean indicating if this is the latest version of the object"),
        @WritesAttribute(attribute = OBS_LAST_MODIFIED, description = "The last modified time in milliseconds since epoch in UTC time"),
        @WritesAttribute(attribute = OBS_LENGTH, description = "The size of the object in bytes"),
        @WritesAttribute(attribute = OBS_STORAGE_CLASS, description = "The storage class of the object"),
        @WritesAttribute(attribute = OBS_VERSION, description = "The version of the object, if applicable"),
        @WritesAttribute(attribute = OBS_USER_META + ".__", description = "If 'Write User Metadata' is set to 'True', the user defined metadata associated to the OBS object that is being listed " +
                "will be written as part of the flowfile attributes")})
public class ListOBSObject extends AbstractListHuaweiProcessor {
    public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(Arrays.asList(
            OBS_REGION,
            ENDPOINT_OVERRIDE_URL,
            BUCKET,
            ACCESS_KEY,
            SECRET_KEY,
            HUAWEI_CREDENTIALS_PROVIDER_SERVICE,
            DELIMITER,
            PREFIX,
            LISTING_STRATEGY,
            ListedEntityTracker.TRACKING_STATE_CACHE,
            ListedEntityTracker.TRACKING_TIME_WINDOW,
            ListedEntityTracker.INITIAL_LISTING_TARGET,
            RECORD_WRITER,
            MAX_AGE,
            MIN_AGE,
            MAX_SIZE,
            MIN_SIZE,
            WRITE_USER_METADATA,
            TIMEOUT,
            PROXY_CONFIGURATION_SERVICE
            ));

    public static final Set<Relationship> relationships = Collections.singleton(REL_SUCCESS);

    protected volatile ObsClient client;

    @OnScheduled
    public void onScheduled(final ProcessContext context) {
        client = OBSUtils.createClient(context, getLogger());
    }

    @Override
    protected Map<String, String> createAttributes(ListableEntity entity, ProcessContext context) {
        OBSRecord record = (OBSRecord) entity;
        VersionOrDeleteMarker versionSummary = record.getVersionSummary();
        if (versionSummary == null) {
            return new HashMap<>();
        }
        final Map<String, String> attributes = new HashMap<>();
        attributes.put(CoreAttributes.FILENAME.key(), versionSummary.getObjectKey());
        attributes.put(OBS_BUCKET, versionSummary.getBucketName());
        if (versionSummary.getOwner() != null) {
            attributes.put(OBS_OWNER, versionSummary.getOwner().getId());
        }
        attributes.put(OBS_E_TAG, versionSummary.getEtag());
        attributes.put(OBS_LAST_MODIFIED, String.valueOf(versionSummary.getLastModified().getTime()));
        attributes.put(OBS_LENGTH, String.valueOf(versionSummary.getSize()));
        attributes.put(OBS_STORAGE_CLASS, versionSummary.getStorageClass());
        attributes.put(OBS_IS_LATEST, String.valueOf(versionSummary.isLatest()));
        if (versionSummary.getVersionId() != null) {
            attributes.put(OBS_VERSION, versionSummary.getVersionId());
        }
        // Add user-defined metadata
        if (record.getObjectMetadata() != null) {
            for (Map.Entry<String, Object> e : record.getObjectMetadata().getAllMetadata().entrySet()) {
                attributes.put(OBS_USER_META + "." + e.getKey(), (String) e.getValue());
            }
        }
        return attributes;
    }

    @Override
    protected String getPath(ProcessContext context) {
        return context.getProperty(BUCKET).evaluateAttributeExpressions().getValue();
    }

    @Override
    protected List performListing(ProcessContext context, Long minTimestamp, ListingMode listingMode) throws IOException {
        final long minimumTimestamp = minTimestamp == null ? 0 : minTimestamp;
        final String region = context.getProperty(OBS_REGION).getValue();
        final String endpoint = context.getProperty(ENDPOINT_OVERRIDE_URL).getValue();
        List<OBSRecord> list = new ArrayList<>();
        int batchCount = 0;
        ObsBucketLister bucketLister = new OBSObjectBucketLister(client, context);
        do {
            ListVersionsResult versionListing = bucketLister.listVersions();

            for (VersionOrDeleteMarker versionSummary : versionListing.getVersions()) {
                if (!isFileInfoMatchesWithAgeAndSize(context, minimumTimestamp, versionSummary.getLastModified().getTime(), versionSummary.getSize())) {
                    continue;
                }
                ObjectMetadata objectMetadata = getObjectMetadata(context, client, versionSummary);
                OBSRecord record = new OBSRecord();
                record.setObjectMetadata(objectMetadata);
                record.setRegion(region);
                record.setEndpoint(endpoint);
                record.setVersionSummary(versionSummary);
                list.add(record);
                batchCount++;
            }
            getLogger().warn("ListVersionsResult count : {}<filtered>/{}<All>", batchCount, versionListing.getVersions().length);
            bucketLister.setNextMarker();
            getLogger().info("Successfully listed {} new files from OBS; routing to success", batchCount);
            batchCount = 0;
            // 是否完成
        } while (bucketLister.isTruncated());
        return list;
    }

    protected boolean isFileInfoMatchesWithAgeAndSize(final ProcessContext context, final long minimumTimestamp,
                                                      final long lastModified, final long size) {
        final long minSize = context.getProperty(MIN_SIZE).asDataSize(DataUnit.B).longValue();
        final Double maxSize = context.getProperty(MAX_SIZE).asDataSize(DataUnit.B);
        final long minAge = context.getProperty(MIN_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
        final Long maxAge = context.getProperty(MAX_AGE).asTimePeriod(TimeUnit.MILLISECONDS);

        if (lastModified < minimumTimestamp) {
            return false;
        }
        final long fileAge = System.currentTimeMillis() - lastModified;
        if (minAge > fileAge) {
            return false;
        }
        if (maxAge != null && maxAge < fileAge) {
            return false;
        }
        if (minSize > size) {
            return false;
        }
        return maxSize == null || maxSize >= size;
    }

    @Override
    protected boolean isListingResetNecessary(PropertyDescriptor property) {
        return BUCKET.equals(property)
                || OBS_REGION.equals(property);
    }

    @Override
    protected Scope getStateScope(PropertyContext context) {
        return null;
    }

    @Override
    protected RecordSchema getRecordSchema() {
        return null;
    }

    @Override
    protected Integer countUnfilteredListing(ProcessContext context) throws IOException {
        return null;
    }

    @Override
    protected String getListingContainerName(ProcessContext context) {
        return null;
    }

    @Override
    protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return properties;
    }

    @Override
    public Set<Relationship> getRelationships() {
        return relationships;
    }

    private ObjectMetadata getObjectMetadata(ProcessContext context, ObsClient client, VersionOrDeleteMarker versionSummary) {
        ObjectMetadata objectMetadata = null;
        if (context.getProperty(WRITE_USER_METADATA).asBoolean()) {
            try {
                GetObjectMetadataRequest request = new GetObjectMetadataRequest(versionSummary.getBucketName(), versionSummary.getKey());
                objectMetadata = client.getObjectMetadata(request);
            } catch (final Exception e) {
                getLogger().warn("Failed to obtain User Metadata for Obs Object {} in bucket {}. Will list Obs Object without the user metadata",
                        new Object[]{versionSummary.getKey(), versionSummary.getBucketName()}, e);
            }
        }
        return objectMetadata;
    }
}

